package com.flink.batch.core;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.UnionOperator;
import org.apache.flink.api.java.operators.join.JoinOperatorSetsBase;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;

import java.util.ArrayList;
import java.util.List;

/**
 * @version v1.0
 * @ProjectName flinkwarehouse
 * @ClassName UnionDemo
 * @Description TODO Union 两个数组的合并实例
 * @Author wudl
 * @Date 2020/8/28 15:16
 */

public class UnionDemo {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        List<Tuple2<Integer,String>> list1 = new ArrayList<>();
        list1.add(new Tuple2<>(1,"hdfs"));
        list1.add(new Tuple2<>(2,"spark"));
        list1.add(new Tuple2<>(3,"flink"));
        list1.add(new Tuple2<>(4,"hive"));
        list1.add(new Tuple2<>(5,"clinkhouse"));

        List<Tuple2<Integer,String>> list2 = new ArrayList<>();
        list2.add(new Tuple2<>(1,"java"));
        list2.add(new Tuple2<>(2,"Python"));

        DataSet<Tuple2<Integer, String>> ds1 = env.fromCollection(list1);
        DataSet<Tuple2<Integer, String>> ds2 = env.fromCollection(list2);
        DataSet<Tuple2<Integer, String>> union = ds1.union(ds2);

        /**
         *集合进行做关联就是说id一样的 以左边的集合为准， 没有就为空
         */

         ds1.leftOuterJoin(ds2).where(0).equalTo(0).with(new JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, Object>() {
             @Override
             public Object join(Tuple2<Integer, String> first, Tuple2<Integer, String> second) throws Exception {

                 System.out.println("*****"+first);
                 System.out.println("****2222*"+second);

                 if(second == null){
                           return new Tuple3<>(first.f0,first.f1,"null");
                       }else{
                           return new Tuple3<>(first.f0,first.f1,second.f1);
                       }
             }
         }).print();
        System.out.println("************************************************************");
        union.print();

    }
}
